package controller

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/go-redis/redis/v8"
	"github.com/gorilla/websocket"
	"gopkg.in/fatih/set.v0"
	"log"
	"my_chat/dao/mysql"
	"my_chat/dao/redis_m"
	"my_chat/logic"
	"my_chat/model"
	"sync"
)

// 消息类型
const (
	CMD_SINGLE_MSG = 10 // 单聊
	CMD_ROOM_MSG   = 11 //群聊
	CMD_HEART      = 0  // 心跳
)

// 在线用户-websocket 映射关系
var syncMap = sync.Map{}

//读写锁
var rwlocker sync.RWMutex

// 存储用户websocket连接及其他信息
type Node struct {
	Conn *websocket.Conn
	//并行转串行,
	DataQueue chan []byte
	GroupSets set.Interface // 集合
}

func Chat(c *gin.Context) {
	//升级get请求为webSocket协议
	conn, err := (&websocket.Upgrader{}).Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		fmt.Println("连接错误")
		return
	}

	node := &Node{
		Conn:      conn,
		DataQueue: make(chan []byte, 50),
		GroupSets: set.New(set.ThreadSafe), // 线程安全的集合
	}

	userId, ok := c.Get("userId")

	if !ok {
		// 当前登入用户无效，跳转到登入页面
		fmt.Println("无法获取用户ID， 跳转到登入页面重新登入")
		Redirect(c)
		return
	}

	//获取用户全部群Id， 保存到node中
	comIds, err := logic.GetUserComids(userId.(int64))
	if err != nil {
		fmt.Println("获取用户群错误")
		return
	}
	for _, v := range comIds {
		node.GroupSets.Add(v)
	}

	syncMap.Store(userId.(int64), node)

	// 发送消息
	go sendData(node)

	// 接受消息
	go recvData(node)
	sendMsg(userId.(int64), []byte("hello,world!")) // 第一次连接发送hello
}

//ws发送协程
func sendData(node *Node) {
	for {
		select {
		case data := <-node.DataQueue:
			err := node.Conn.WriteMessage(websocket.TextMessage, data)
			if err != nil {
				fmt.Println(err.Error())
				return
			}
		}
	}
}

// ws接受消息
func recvData(node *Node) {
	for {
		_, data, err := node.Conn.ReadMessage()
		// 心跳检测
		heart := string(data)
		if heart == "heart" {
			fmt.Println("心跳检测")
			continue
		}
		if err != nil {
			fmt.Println(err.Error())
			return
		}
		if err := logic.SaveMessage(data); err != nil {
			fmt.Println(err.Error())
			return
		}
		str := string(data[:])
		handle(str)
		fmt.Printf("[ws]<=%s\n", data)

	}
}

func handle(str string) {
	msgChan <- str
}

var msgChan chan string = make(chan string, 1024)

// redis消息订阅发布
// 发布
func pub(ctx context.Context, rdb *redis.Client) {
	for {
		select {
		case data := <-msgChan:
			// Publish a message.
			err := rdb.Publish(ctx, "my_chat", data).Err()
			if err != nil {
				fmt.Println(err.Error())
				return
			}
		}
	}
}

// 订阅
func sub(ctx context.Context, rdb *redis.Client) {
	pubsub := rdb.Subscribe(ctx, "my_chat")

	// Wait for confirmation that subscription is created before publishing anything.
	_, err := pubsub.Receive(ctx)
	if err != nil {
		panic(err)
	}

	// Go channel which receives messages.
	ch := pubsub.Channel()

	// Consume messages.
	for msg := range ch {
		go dispatch([]byte(msg.Payload))
	}
}

func InitMq() {
	rdb := redis_m.Client
	ctx := context.Background()
	go sub(ctx, rdb)
	go pub(ctx, rdb)
	fmt.Println("初始化聊天消息队列服务")
}

//后端调度逻辑处理
func dispatch(data []byte) {
	//解析data为message
	msg := model.Message{}
	err := json.Unmarshal(data, &msg)
	if err != nil {
		log.Println(err.Error())
		return
	}
	//根据cmd对逻辑进行处理
	switch msg.Cmd {
	case CMD_SINGLE_MSG:
		// 单聊转发逻辑
		sendMsg(msg.Dstid, data)
	case CMD_ROOM_MSG:
		//群聊转发逻辑
		userIds, err := mysql.GroupUsers(msg.Dstid)
		if err != nil {
			fmt.Println("获取群内用户出错")
			return
		}
		for _, v := range userIds {
			if msg.Userid != v {
				sendMsg(v, data)
			}
		}
		//syncMap.Range(func(userId, node interface{}) bool {
		//	if node.(*Node).GroupSets.Has(msg.Dstid) {
		//		//自己排除,不发送
		//		if msg.Userid != userId.(int64) {
		//			node.(*Node).DataQueue <- data
		//		}
		//
		//	}
		//	return true
		//})

	case CMD_HEART:
		//一般啥都不做
	}
}

//todo 发送消息
func sendMsg(userId int64, msg []byte) {
	node, ok := syncMap.Load(userId)
	if ok {
		node.(*Node).DataQueue <- msg
	}
}

// 查找用户聊天信息
func GetMessages(c *gin.Context) {
	p := new(model.ContactValid)
	if err := c.ShouldBindJSON(p); err != nil {
		// 请求参数有误，直接返回响应
		fmt.Println(err)
		RespFail(c, err.Error())
		return
	}

	if p.Dstobj == 0 {
		err := errors.New("请传入对面id")
		RespFail(c, err.Error())
		return
	}
	data, err := logic.GetMessages(p)
	if err != nil {
		fmt.Println(err)
		RespFail(c, err.Error())
		return
	} else {
		RespOkList(c, data, len(data.(model.Messages)))
	}
}
